/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.di.cst.lls.common;

import cn.ac.iie.di.cst.lls.conf.Configuration;
import cn.ac.iie.di.cst.lls.conf.ConfigurationException;
import cn.ac.iie.di.cst.lls.se.CstdxStartup;
import static cn.ac.iie.di.cst.lls.util.MiscTools.getIpAddress;
import java.io.File;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.avro.Protocol;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericDatumReader;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.io.DatumReader;
import org.apache.log4j.Logger;

/**
 *
 * @author carlisle lee
 * @date: 2014-11-05
 */
public class RuntimeEnv {

    private static Configuration conf = null;
    private static Map<String, Object> dynamicParams = new HashMap<String, Object>();

    public static String ConsumerNamesrv = "consumerNamesrv";
    public static String ProducerNamesrv = "producerNamesrv";
    public static String ChangeColumns = "changeColumns";
    public static String MqList = "mqList";
    public static String ProductConsumersname = "productConsumersname";
    public static final String HostNameIp = "hostNameIp";

    public static ConcurrentHashMap<String, BlockingQueue> mq2queue = new ConcurrentHashMap<String, BlockingQueue>();
    public static ConcurrentHashMap<String, DatumReader<GenericRecord>> mqdocSchema = new ConcurrentHashMap<String, DatumReader<GenericRecord>>();
    public static Schema docsSchema = null;
    public static GenericDatumReader<GenericRecord> docsReader = null;
    public static HashMap<Schema, DatumReader<GenericRecord>> schemaDatumReader = new HashMap<Schema, DatumReader<GenericRecord>>();
    public static HashMap<String, Schema> mqschema = new HashMap<String, Schema>();
    public static String[] changeColumns = null;
    public static String BatchSize = "batchSize";
    public static final String ThSize = "thSize";
     public static AtomicBoolean isShouldExit = new AtomicBoolean(false);

    public static int SendMessageSize;

    private static org.apache.log4j.Logger logger = Logger.getLogger(RuntimeEnv.class.getName());

    public static boolean initialize() throws ConfigurationException {
        String cp = "cst-lls2ggcc.properties";
        try {
            conf = Configuration.getConfiguration(cp);
        } catch (Exception E) {
            return false;
        }
        String hostnameip = getIpAddress();
        dynamicParams.put(HostNameIp, hostnameip);
        logger.info("hostNameIp rocketme instance " + hostnameip);

        getConfAndSaveAsInteger(BatchSize);
        getConfAndSaveAsInteger(ThSize);
        
        
        String consumerNamesrv = conf.getString(ConsumerNamesrv, "");
        if (consumerNamesrv.isEmpty()) {
            logger.error("parameter consumerNamesrv does not exist or is not defined");
            return false;
        }
        dynamicParams.put(ConsumerNamesrv, consumerNamesrv);

        String producerNamesrv = conf.getString(ProducerNamesrv, "");
        if (producerNamesrv.isEmpty()) {
            logger.error("parameter producerNamesrv does not exist or is not defined");
            return false;
        }
        dynamicParams.put(ProducerNamesrv, producerNamesrv);

        String mqList = conf.getString(MqList, "");
        if (mqList.isEmpty()) {
            logger.error("parameter producerNamesrv does not exist or is not defined");
            return false;
        }
        dynamicParams.put(MqList, mqList);

        String productConsumersname = conf.getString(ProductConsumersname, "");
        if (mqList.isEmpty()) {
            logger.error("parameter ProductConsumersname  does not exist or is not defined");
            return false;
        }
        dynamicParams.put(ProductConsumersname, productConsumersname);

        String sendMessageSize = conf.getString("sendMessageSize", "");
        if (sendMessageSize.isEmpty()) {
            logger.error("parameter sendMessageSize  does not exist or is not defined");
            return false;
        }
        SendMessageSize = Integer.valueOf(sendMessageSize);

        String[] mqsList = mqList.split("[|]");
        for (int k = 0; k < mqsList.length; k++) {
            BlockingQueue<GenericRecord> tableque = new ArrayBlockingQueue<GenericRecord>(10000);
            mq2queue.put(mqsList[k], tableque);
        }

        String changecolumns = conf.getString(ChangeColumns, "");

        if (changecolumns.isEmpty()) {
            logger.error("parameter  changeColumns does not exist or is not defined");
            return false;
        }
        changeColumns = changecolumns.split("[|]");

        File docsSchemaFile = new File("docs.json");
        try {
            Protocol protocol = Protocol.parse(docsSchemaFile);
            docsSchema = protocol.getType("docs");

        } catch (Exception E) {
            System.out.println("找不到docs.json" + E);
        }
        docsReader = new GenericDatumReader<GenericRecord>(docsSchema);
        for (int k = 0; k < mqsList.length; k++) {
            File docSche = new File(mqsList[k] + ".json");
            Schema docSchema = null;
            try {
                Protocol protocol = Protocol.parse(docSche);
                docSchema = protocol.getType(mqsList[k]);
                GenericDatumReader docReader = new GenericDatumReader<GenericRecord>(docSchema);
                schemaDatumReader.put(docSchema, docReader);
                mqschema.put(mqsList[k], docSchema);

            } catch (Exception E) {
                System.out.println("找不到docs.json" + E);
            }
        }
        return true;
    }

    public static void addParam(String pParamName, Object pValue) {
        synchronized (dynamicParams) {
            dynamicParams.put(pParamName, pValue);
        }
    }

    public static Object getParam(String pParamName) {
        return dynamicParams.get(pParamName);
    }

    public static boolean getConfAndSaveAsString(String confName) {
        String tmp = conf.getString(confName, "");
        if (tmp.isEmpty()) {
            logger.error("parameter" + confName + "does not exist or is not defined");
            return false;
        }
        dynamicParams.put(confName, tmp);
        logger.info(confName + ":" + tmp);
        return true;
    }

    public static boolean getConfAndSaveAsInteger(String confName) {
        String tmp = conf.getString(confName, "");
        if (tmp.isEmpty()) {
            logger.error("parameter " + confName + " does not exist or is not defined");
            return false;
        }
        try {
            dynamicParams.put(confName, Integer.parseInt(tmp));
            logger.info(confName + ":" + tmp);
        } catch (Exception ex) {
            logger.error("parameter " + confName + " 's format is wrong for " + ex.getMessage(), ex);
            return false;
        }
        return true;
    }
}
